Go 并发工具选型：errgroup、singleflight、semaphore 各管哪一段

并发代码里有三种容易混淆的需求：一批任务一起成功或失败；同一个数据正在加载时别重复加载；资源不够时先排队。errgroup、singleflight、semaphore 分别解决它们。选型时先判断当前问题属于哪一种，比一开始就把三个工具叠起来更容易得到清楚的生命周期。

本文标准库使用 Go 1.27.2，三个扩展包都固定为 golang.org/x/sync v0.16.0。这是一个独立发布的 Go 官方扩展模块，不是标准库的一部分；升级工具链不会自动升级项目里的 x/sync。本文讨论与测试均对应这个固定版本，不把 pkg.go.dev 上未来的默认版本行为混进来。^[1]

用一个问题选出入口

你遇到的问题	先看哪个工具	它协调的单位	它不做什么
同一个请求拆成多项工作，一项失败就通知其他项停止	errgroup	一组函数调用	不强杀任务，不自动汇总所有错误
很多请求同时加载同一个 key，希望只执行一份加载	singleflight	同一 Group、同一 key 的重叠调用	不保留跨时间的缓存，不为你设计共享任务的取消
同时只能用 N 个连接，或最多占某个资源预算	semaphore	正在占用的权重总和	不限制每秒次数，不收集任务错误
只想等几条 goroutine 结束	标准库 WaitGroup	任务计数	不必为了没有的错误需求引入扩展包

这些工具可以组合，但每多一层等待，就多一处要解释“谁取消它、谁最后等它结束”。例如缓存 miss 先 singleflight 合并，再在共享加载函数内申请 semaphore 名额，可以让一个 key 的重复请求不重复占下游名额；这仍需要缓存、key 隔离和独立的加载超时设计。

errgroup：共同工作的错误边界

常用 API 只有五个：

API	行为	使用时最容易漏掉的条件
`WithContext(parent)`	返回 Group 和派生 context	任务必须使用返回的 ctx，才会观察到同组失败
`g.Go(func() error)`	登记并启动任务	受 SetLimit 限制时，提交本身可能阻塞
`g.Wait()`	等全部任务返回，返回首个非 nil error	首个错误并不让 Wait 立刻返回
`g.SetLimit(n)`	限同时活跃的任务数	活跃任务存在时不能修改 limit；0 禁止新增，负数不限制
`g.TryGo(f)`	有空位才启动，返回是否启动	false 需要调用方决定拒绝、稍后重试还是别的处理

零值 errgroup.Group 可以直接使用，但没有数量限制，也没有由 WithContext 建立的共同取消。Group 用于一项整体工作，不应复用成不同任务之间的共享管理器。^[2]

一个分支失败，另一个怎样退出

g, ctx := errgroup.WithContext(context.Background())
started := make(chan struct{})
g.Go(func() error { close(started); <-ctx.Done(); return ctx.Err() })
g.Go(func() error { <-started; return errors.New("upstream failed") })
return g.Wait()

本次输出：

text

errgroup: upstream failed

第一个任务先进入等待，第二个任务返回业务错误，group 记录错误并取消派生 ctx。第一个任务收到取消后返回，Wait 才能完成。如果第一个任务忽略 ctx，卡在一个永不返回的调用里，Wait 也会一直等。错误传播和强制结束执行不是同一回事。^[2:1]

Wait() 返回时派生 ctx 也会被取消，哪怕所有任务都成功。因此不能把这个 ctx 当成“成功后下一阶段继续使用的上下文”。成功后继续工作，应回到仍有效的父级或明确创建新的阶段 context。本文有一个测试，任务返回 nil、Wait 返回 nil，但派生 ctx 的 Err 已是 context.Canceled。^[2:2]

反过来，父级取消也不意味着 Wait 一定返回 context.Canceled。Wait 返回的是任务函数的错误；如果所有函数都返回 nil，或者组里根本没有返回错误的任务，Wait 可以是 nil。希望取消体现为返回错误，任务需要按照接口约定检查 ctx 并返回相应错误。^[2:3]

限数量要同时看“提交者”会不会等

本文另一个小实验先设置 SetLimit(1)，启动一个等 release 的任务，再调用 TryGo。第一个任务还占着名额时，本次输出：

text

trygo at capacity: false

这能直接表达拒绝立即入组。换成 Go，调用方会等待名额，而这次内部等待没有 ctx 参数；取消 group 的 ctx 不会直接把一个卡在 Go 入组动作上的调用变成返回错误。若需要排队也能取消，可以使用显式 worker 队列或 semaphore.Acquire。^[2:4]

也不要在一个已经占用全部名额的组内同步提交子任务并等待它。极端例子是 limit=1，父任务自己占着唯一名额，又调用同组 Go 启动子任务：父任务等空位，空位却要等父任务返回，形成死锁。

errgroup v0.16.0 不把 panic 或 runtime.Goexit 改造成普通 error。预期业务失败用 error 表达；真正需要恢复 panic 的场景，应在合适的任务边界明确设计，不能假设 Wait 会收到它。^[3]

singleflight：只合并正在重叠的同 key 调用

一个热门缓存 key 刚好过期，同时来了 100 个请求。每个人都发现缓存没有值，随后各自查一次数据库，会把一次 miss 放大为 100 次加载。singleflight 可以让这些请求等待同一份正在进行的加载。^[4]

API	返回方式	取消与生命周期
`Do(key, fn)`	同步返回值、error、shared	调用方直接等待；API 没有 context 参数
`DoChan(key, fn)`	返回一个将接收 `Result` 的只读 channel	可用 select 放弃等待；不会因此取消 fn
`Forget(key)`	不等待结果	后续调用不再等旧调用；旧调用仍会运行

同一 key 的重叠调用共享首个调用传入函数的结果和错误，后续请求传入的函数不会另外运行。Shared 表示该结果是否给了多个调用方，不是“当前调用是不是发起者”的标志。最初启动加载的那个人也可能拿到 Shared=true。^[4:1]

用一个开关保证两次请求真的重叠

gate := make(chan struct{})
fn := func() (any, error) {
    calls.Add(1)
    <-gate
    return "value", nil
}
a := g.DoChan("key", fn)
b := g.DoChan("key", fn)
close(gate)
x, y := <-a, <-b

gate 关闭前加载不会完成，因此第二次调用到来时第一份工作一定还在。本次完整程序输出：

text

singleflight calls: 1 shared: true

不要用 time.Sleep(10*time.Millisecond) 猜测两次调用是否重叠；那会把示例是否成立交给调度和机器负载。这个小屏障直接表达了实验条件。

如果依次调用两次 Do，等第一次结束再发第二次，结果就不同：

g.Do("key", f)
g.Do("key", f)

本次输出：

text

sequential calls: 2

singleflight 在调用结束后不会继续保存该结果。需要缓存时仍要维护缓存及过期策略。常见结构是先查缓存，miss 后进入 singleflight，再在共享函数中复查缓存，最后加载并写回；第二次检查可避免等待期间已有别的路径填入结果后仍重复加载。这是缓存协议的设计建议，不是 singleflight 隐含执行的动作。

放弃等待，和取消共享工作，要分别决定

正在绘制图示…

查看图源代码

sequenceDiagram
    participant A as 请求 A
    participant G as singleflight Group
    participant F as 共享加载函数
    participant B as 请求 B
    A->>G: DoChan(key, fn)
    G->>F: 启动一次 fn
    B->>G: DoChan(同 key, fn)
    A->>A: 自己的 ctx 取消，停止等结果
    Note over G,F: A 放弃等待不会自动停止 fn
    F-->>G: 返回 value 或 error
    G-->>B: 交付共享结果

原始尺寸下可横向或纵向滚动；按 Escape 关闭。

正在绘制图示…

查看图源代码

sequenceDiagram
    participant A as 请求 A
    participant G as singleflight Group
    participant F as 共享加载函数
    participant B as 请求 B
    A->>G: DoChan(key, fn)
    G->>F: 启动一次 fn
    B->>G: DoChan(同 key, fn)
    A->>A: 自己的 ctx 取消，停止等结果
    Note over G,F: A 放弃等待不会自动停止 fn
    F-->>G: 返回 value 或 error
    G-->>B: 交付共享结果

DoChan 在 fn 正常返回时向结果 channel 发送一项 Result，之后不关闭 channel。因此读取一次即可，不能用 for range resultCh 等它结束。调用方可以 select 自己的 Done 与结果 channel，让自己的请求及时返回，但这仅取消等待，不取消共享函数。^[4:2]

v0.16.0 还有一条异常边界：fn 的 panic 不会作为普通 Result.Err 交付；fn 调用 runtime.Goexit 可使 DoChan 的 channel 得不到结果。共享函数应正常返回值或 error，等待方也应有自己的取消或超时；这不能把 panic 变成普通错误。这两种异常行为仅按该版本 doCall 源码核对，本文没有执行对应失败实验。^[4:3]

共享函数用哪个 context，需要结合任务所有权决定。直接捕获第一个请求的 ctx，那个请求离开时可能让所有共享等待者一起失败；完全改成没有截止时间的 Background，又可能留下无人管理的长期工作。一种可考虑的设计是由服务持有共享加载的生命周期，给每次加载独立的合理超时，请求只决定自己是否继续等。这并不是默认答案：加载是否有副作用、是否必须随发起者终止，都要明确约定。

key 也是隔离边界。会影响结果的租户、权限、参数和数据版本必须纳入 key，或用不同 Group 隔离；只用一个模糊的资源 ID，可能让不该共享的请求拿到同一结果。返回值若是可变指针、slice 或 map，多个人收到的仍可能是同一个对象，应约定只读或复制，而不是以为“共享加载”自带深拷贝。^[4:4]^[5]

Forget 不是取消。旧调用仍会给旧等待者交付结果，新调用却可以重新启动，所以同一 key 可能短时间内有两份工作重叠。只有接受这个后果时才调用它；不要把它当作安全终止一个卡住加载的接口。^[4:5]

semaphore：用权重表示资源占用

标准 channel 很适合每个任务占一个名额的情况。semaphore.Weighted 则允许一项任务占多个单位：例如总预算是 10，轻任务需要 1，重任务需要 4，让活跃任务权重之和不超过 10。^[6]

API	做什么	返回后该做什么
`NewWeighted(limit)`	建立总容量	业务层先校验 limit 为正
`Acquire(ctx, weight)`	等到足够资源或 ctx 取消	成功才获得名额；失败返回 ctx.Err，不能 Release
`TryAcquire(weight)`	不等待，只尝试获取	返回 true 才获得名额
`Release(weight)`	归还占用	与成功获取的权重配对，多归还会 panic

最重要的代码位置是“先 Acquire，再 go”：

if err := sem.Acquire(ctx, n); err != nil {
    acquireErr = err
    break
}
wg.Add(1)
go func(weight int64) {
    defer wg.Done()
    defer sem.Release(weight)
    // 实际工作
}(n)

如果反过来，每次先启动 goroutine，再让它在 Acquire 排队，只限制了获准执行者，没限制排队 goroutine 的数量。输入能无限增长时，依然会堆积大量等待者。固定 worker pool 或在提交线程中等名额，更容易让总量有界。

文末 WeightedWork 把权重 2、1、2、1 放进容量为 3 的信号量，并用原子计数观察活跃权重峰值。本次输出：

text

semaphore within limit: true <nil>

程序没有报告“峰值必然达到 3”：任务太短，实际重叠取决于调度，测试只检查观察到的峰值在合法范围内。该程序只验证获取、释放、取消和收尾，不模拟真实网络耗时。

名额获取成功后，取消不会替你归还

Acquire 的 ctx 用来控制等名额。成功之后，任务拥有名额，必须在结束时 Release；ctx 稍后取消，不会自动从仍运行的任务手里抢走名额。任务主体也必须独立响应 ctx。^[6:1]

假设提交第 4 项时 Acquire 被取消，第 1 至第 3 项可能已经启动。函数不能只返回获取错误，把前面的任务丢在后台。本文实现会先跳出提交循环，再 wg.Wait() 等已经启动的工作结束，最后返回获取错误。真实工作若可能失败，还需要用 errgroup 或单独的错误通道收集其错误。

调用前校验 0 < weight <= limit 很重要。v0.16.0 对申请超过总容量的 Acquire 不会立即返回“参数超限”；它等待 ctx 结束。若用 Background，就可能永远等下去。非正权重也不应让业务传入，因为这个 API 不替应用建立完整的资源参数约束。本文先验证整批权重，确保错误输入不会留下已经启动的部分工作。^[7]

该版本的等待队列会顾及排在前面的申请：如果前面的大权重暂时无法满足，不让后面的小请求一直抢走零碎空位，以免大请求长期等不到资源。这可能使有空闲额度时小请求仍等待，是吞吐与避免大请求饥饿之间的实现取舍。它不构成跨调度、跨取消和任意负载的“全局公平”或有限等待时间保证。^[7:1]

组合之前，先写清三个所有者

问题	在常见组合里由谁回答
谁决定本次请求不再等待？	请求自己的 context
谁决定共享加载何时终止？	加载工作的所有者及其 deadline；不能由等待者数量凭空决定
谁确认全部已启动工作结束？	errgroup.Wait、WaitGroup.Wait 或明确的完成信号

例如“同 key 去重，再限制数据库并发”可以写成 singleflight 的 fn 内 Acquire，成功后 defer Release；但不要让每个重复等待者都先占数据库名额，否则名额被没有实际查询的请求占住了。这个建议的前提是名额衡量的是实际数据库工作；如果限制的是入口请求占用，获取位置又会不同。

工具的语义决定哪些操作能连起来，业务所有权决定应该怎样连。能明确说出取消哪一层、等待哪一层、资源在哪一层释放，才算选完工具。

依据与本次验证

Go 1.27.2、Linux/amd64，x/sync v0.16.0。实际运行了首错取消、TryGo 满额拒绝、重叠调用共享、顺序调用不缓存、加权并发限制。测试还覆盖成功 Wait 也取消派生 ctx、放弃等待不停止共享函数、DoChan 不关闭结果 channel、预先取消的 Acquire 与非法权重。go test -race -count=1 ./...、go vet ./... 通过。没有性能基准，没有真实数据库调用，也没有验证其他 x/sync 版本与此完全相同。

完整代码与复现方法

以同一目录为根，按下面标出的相对路径保存文件。使用 Go 1.27.2 运行：

go run .
go test -race -count=1 ./...
go vet ./...

依赖固定为 golang.org/x/sync v0.16.0。首次运行需要该版本已在模块缓存中，或能够访问 Go 模块代理；它是独立模块，不随 Go 1.27.2 自动更新。

go.mod

module example.com/go-articles/08_extensions

go 1.25.0

require golang.org/x/sync v0.16.0

go.sum

text

golang.org/x/sync v0.16.0 h1:ycBJEhp9p4vXvUZNszeOq0kGTPghopOL8q0fq3vstxw=
golang.org/x/sync v0.16.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=

main.go

代码块 · 98 行 · 点击收起 / 展开

package main

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"

	"golang.org/x/sync/errgroup"
	"golang.org/x/sync/semaphore"
	"golang.org/x/sync/singleflight"
)

func GroupFailure() error {
	g, ctx := errgroup.WithContext(context.Background())
	started := make(chan struct{})
	g.Go(func() error { close(started); <-ctx.Done(); return ctx.Err() })
	g.Go(func() error { <-started; return errors.New("upstream failed") })
	return g.Wait()
}
func SharedLoad() (int64, bool) {
	var g singleflight.Group
	var calls atomic.Int64
	gate := make(chan struct{})
	fn := func() (any, error) { calls.Add(1); <-gate; return "value", nil }
	a := g.DoChan("key", fn)
	b := g.DoChan("key", fn)
	close(gate)
	x, y := <-a, <-b
	return calls.Load(), x.Shared && y.Shared && x.Err == nil && y.Err == nil
}
func WeightedWork(ctx context.Context, limit int64, weights []int64) (int64, error) {
	if limit <= 0 {
		return 0, errors.New("limit must be positive")
	}
	// Validate before launching any work.
	for _, n := range weights {
		if n <= 0 || n > limit {
			return 0, errors.New("invalid weight")
		}
	}
	sem := semaphore.NewWeighted(limit)
	var wg sync.WaitGroup
	var current, peak atomic.Int64
	var acquireErr error
	for _, n := range weights {
		if err := sem.Acquire(ctx, n); err != nil {
			acquireErr = err
			break
		}
		wg.Add(1)
		go func(weight int64) {
			defer wg.Done()
			defer sem.Release(weight)
			now := current.Add(weight)
			for {
				old := peak.Load()
				if now <= old || peak.CompareAndSwap(old, now) {
					break
				}
			}
			// Real work belongs here and must honor ctx independently.
			current.Add(-weight)
		}(n)
	}
	wg.Wait()
	return peak.Load(), acquireErr
}
func NotACache() int {
	var g singleflight.Group
	calls := 0
	f := func() (any, error) { calls++; return "value", nil }
	g.Do("key", f)
	g.Do("key", f)
	return calls
}
func TryGoAtCapacity() bool {
	var g errgroup.Group
	g.SetLimit(1)
	release := make(chan struct{})
	g.Go(func() error { <-release; return nil })
	started := g.TryGo(func() error { return nil })
	close(release)
	if err := g.Wait(); err != nil {
		panic(err)
	}
	return started
}
func main() {
	fmt.Println("errgroup:", GroupFailure())
	fmt.Println("trygo at capacity:", TryGoAtCapacity())
	calls, shared := SharedLoad()
	fmt.Println("singleflight calls:", calls, "shared:", shared)
	fmt.Println("sequential calls:", NotACache())
	peak, err := WeightedWork(context.Background(), 3, []int64{2, 1, 2, 1})
	fmt.Println("semaphore within limit:", peak <= 3, err)
}

main_test.go

代码块 · 84 行 · 点击收起 / 展开

package main

import (
	"context"
	"errors"
	"golang.org/x/sync/errgroup"
	"golang.org/x/sync/singleflight"
	"testing"
)

func TestGroupFailure(t *testing.T) {
	if err := GroupFailure(); err == nil || err.Error() != "upstream failed" {
		t.Fatal(err)
	}
}
func TestSingleflight(t *testing.T) {
	if calls, shared := SharedLoad(); calls != 1 || !shared {
		t.Fatal(calls, shared)
	}
}
func TestSemaphore(t *testing.T) {
	peak, err := WeightedWork(context.Background(), 3, []int64{2, 1, 2, 1})
	if err != nil || peak < 1 || peak > 3 {
		t.Fatal(peak, err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	if _, err := WeightedWork(ctx, 3, []int64{1}); !errors.Is(err, context.Canceled) {
		t.Fatal(err)
	}
	if _, err := WeightedWork(context.Background(), 3, []int64{4}); err == nil {
		t.Fatal("invalid weight accepted")
	}
}

func TestNotACache(t *testing.T) {
	if n := NotACache(); n != 2 {
		t.Fatal(n)
	}
}
func TestWaitCancelsEvenSuccess(t *testing.T) {
	g, ctx := errgroup.WithContext(context.Background())
	g.Go(func() error { return nil })
	if err := g.Wait(); err != nil {
		t.Fatal(err)
	}
	if !errors.Is(ctx.Err(), context.Canceled) {
		t.Fatal(ctx.Err())
	}
}

func TestTryGoAtCapacity(t *testing.T) {
	if TryGoAtCapacity() {
		t.Fatal("exceeded limit")
	}
}
func TestAbandonWaitDoesNotCancelSharedWork(t *testing.T) {
	var g singleflight.Group
	entered, release := make(chan struct{}), make(chan struct{})
	a := g.DoChan("k", func() (any, error) { close(entered); <-release; return "value", nil })
	<-entered
	b := g.DoChan("k", func() (any, error) { t.Error("duplicate executed"); return nil, nil })
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	select {
	case <-a:
		t.Fatal("work completed before release")
	case <-ctx.Done():
	}
	close(release)
	r := <-b
	if r.Val != "value" || r.Err != nil || !r.Shared {
		t.Fatal(r)
	}
	// Receiving once is valid; the result channel is not closed afterward.
	<-a
	select {
	case _, ok := <-a:
		if !ok {
			t.Fatal("DoChan result channel closed")
		}
	default:
	}
}

Go Modules Reference：模块版本与 require；x/sync v0.16.0 go.mod。 ↩︎
x/sync v0.16.0 errgroup 源码及 API 注释，WithContext、Go、TryGo、SetLimit、Wait。本次还从模块缓存读取了完整 go doc。 ↩︎ ↩︎ ↩︎ ↩︎ ↩︎
同一固定版本的 Go 实现明确说明不把 panic 传回 Wait；Wait 只返回任务记录的 error。参见 errgroup.go。 ↩︎
x/sync v0.16.0 singleflight，Do、DoChan、Forget、Result.Shared，以及 doCall 对 panic、runtime.Goexit 的处理。 ↩︎ ↩︎ ↩︎ ↩︎ ↩︎ ↩︎
Go 规范：Assignments，引用数据的赋值语义。 ↩︎
x/sync v0.16.0 semaphore，Weighted 及公开方法注释。 ↩︎ ↩︎
同一 semaphore.go 固定源码，Acquire 的超容量路径、Release 的超额归还检查、notifyWaiters 的队首策略；这些属于该版本实现细节。 ↩︎ ↩︎

Go 并发工具选型：errgroup、singleflight、semaphore 各管哪一段 ​

用一个问题选出入口 ​

errgroup：共同工作的错误边界 ​

一个分支失败，另一个怎样退出 ​

限数量要同时看“提交者”会不会等 ​

singleflight：只合并正在重叠的同 key 调用 ​

用一个开关保证两次请求真的重叠 ​

放弃等待，和取消共享工作，要分别决定 ​

semaphore：用权重表示资源占用 ​

名额获取成功后，取消不会替你归还 ​

组合之前，先写清三个所有者 ​

依据与本次验证 ​

完整代码与复现方法 ​

go.mod ​

go.sum ​

main.go ​

main_test.go ​

Go 并发工具选型：errgroup、singleflight、semaphore 各管哪一段

用一个问题选出入口

errgroup：共同工作的错误边界

一个分支失败，另一个怎样退出

限数量要同时看“提交者”会不会等

singleflight：只合并正在重叠的同 key 调用

用一个开关保证两次请求真的重叠

放弃等待，和取消共享工作，要分别决定

semaphore：用权重表示资源占用

名额获取成功后，取消不会替你归还

组合之前，先写清三个所有者

依据与本次验证

完整代码与复现方法

go.mod

go.sum

main.go

main_test.go